package com.xyzwps.reactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

public class Ex02Subscriber {

    public static void main(String[] args) {
        Flux<Integer> flux = Flux.range(1, 20);
        flux.subscribe(new TheSubscriber());
    }

    static class TheSubscriber extends BaseSubscriber<Integer> {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            System.out.println("Subscribed");
            request(1);
        }

        @Override
        protected void hookOnNext(Integer value) {
            System.out.println("Value: " + value);
            request(1);
        }

        @Override
        protected void hookOnComplete() {
            System.out.println("Completed.");
        }

        @Override
        protected void hookFinally(SignalType type) {
            System.out.println("Finally " + type);
        }
    }
}
